package com.leansoft.bigqueue;

import java.io.File;
import java.io.IOException;
import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import com.leansoft.bigqueue.page.IMappedPage;
import com.leansoft.bigqueue.page.IMappedPageFactory;
import com.leansoft.bigqueue.page.MappedPageFactoryImpl;
import com.leansoft.bigqueue.utils.Calculator;
import com.leansoft.bigqueue.utils.FileUtil;

/**
 * A big array implementation supporting sequential append and random read.
 * <p>
 * Main features:
 * 1. FAST : close to the speed of direct memory access, extremely fast in append only and sequential read modes,
 * sequential append and read are close to O(1) memory access, random read is close to O(1) memory access if
 * data is in cache and is close to O(1) disk access if data is not in cache.
 * 2. MEMORY-EFFICIENT : automatic paging & swapping algorithm, only most-recently accessed data is kept in memory.
 * 3. THREAD-SAFE : multiple threads can concurrently read/append the array without data corruption.
 * 4. PERSISTENT - all array data is persisted on disk, and is crash resistant.
 * 5. BIG(HUGE) - the total size of the array data is only limited by the available disk space.
 *
 * @author bulldog
 */
public class BigArrayImpl implements IBigArray {

    // folder name for index page 索引页目录
    final static String INDEX_PAGE_FOLDER = "index";
    // folder name for data page 数据页目录
    final static String DATA_PAGE_FOLDER = "data";
    // folder name for meta data page 元数据页目录
    final static String META_DATA_PAGE_FOLDER = "meta_data";

    // 2 ^ 17 = 1024 * 128
    final static int INDEX_ITEMS_PER_PAGE_BITS = 17; // 1024 * 128
    // number of items per page index 每一页有多少个 items
    final static int INDEX_ITEMS_PER_PAGE = 1 << INDEX_ITEMS_PER_PAGE_BITS;
    // 2 ^ 5 = 32
    final static int INDEX_ITEM_LENGTH_BITS = 5;
    // length in bytes of an index item ， index item 长度
    final static int INDEX_ITEM_LENGTH = 1 << INDEX_ITEM_LENGTH_BITS;
    // size in bytes of an index page， index page 大小 = item_length * item_number
    final static int INDEX_PAGE_SIZE = INDEX_ITEM_LENGTH * INDEX_ITEMS_PER_PAGE;

    // size in bytes of a data page 数据页大小
    final int DATA_PAGE_SIZE;

    // default size in bytes of a data page 默认数据页大小 128M
    public final static int DEFAULT_DATA_PAGE_SIZE = 128 * 1024 * 1024;
    // minimum size in bytes of a data page 最小数据页大小 32M
    public final static int MINIMUM_DATA_PAGE_SIZE = 32 * 1024 * 1024;
    // seconds, time to live for index page cached in memory 索引页默认 ttl
    final static int INDEX_PAGE_CACHE_TTL = 1000;
    // seconds, time to live for data page cached in memory 数据页默认 ttl
    final static int DATA_PAGE_CACHE_TTL = 1000;
    // 2 ^ 4 = 16
    final static int META_DATA_ITEM_LENGTH_BITS = 4;
    // size in bytes of a meta data page 元数据页大小 16 字节
    final static int META_DATA_PAGE_SIZE = 1 << META_DATA_ITEM_LENGTH_BITS;

    //	private final static int INDEX_ITEM_DATA_PAGE_INDEX_OFFSET = 0;
    //	private final static int INDEX_ITEM_DATA_ITEM_OFFSET_OFFSET = 8;
    // index item 和 data item 的偏移
    private final static int INDEX_ITEM_DATA_ITEM_LENGTH_OFFSET = 12;
    // timestamp offset of an data item within an index item 时间戳偏移
    final static int INDEX_ITEM_DATA_ITEM_TIMESTAMP_OFFSET = 16;

    // directory to persist array data 数据存储目录
    String arrayDirectory;

    // factory for index page management(acquire, release, cache) 索引页工厂
    IMappedPageFactory indexPageFactory;
    // factory for data page management(acquire, release, cache) 数据页工厂
    IMappedPageFactory dataPageFactory;
    // factory for meta data page management(acquire, release, cache) 元数据页工厂
    IMappedPageFactory metaPageFactory;

    // only use the first page 元数据页索引
    static final long META_DATA_PAGE_INDEX = 0;

    // head index of the big array, this is the read write barrier.
    // readers can only read items before this index, and writes can write this index or after
    final AtomicLong arrayHeadIndex = new AtomicLong();
    // tail index of the big array,
    // readers can't read items before this tail
    final AtomicLong arrayTailIndex = new AtomicLong();

    // head index of the data page, this is the to be appended data page index
    long headDataPageIndex;
    // head offset of the data page, this is the to be appended data offset
    int headDataItemOffset;

    // lock for appending state management
    final Lock appendLock = new ReentrantLock();

    // global lock for array read and write management
    final ReadWriteLock arrayReadWritelock = new ReentrantReadWriteLock();
    final Lock arrayReadLock = arrayReadWritelock.readLock();
    final Lock arrayWriteLock = arrayReadWritelock.writeLock();

    /**
     * A big array implementation supporting sequential write and random read,
     * use default back data file size per page, see {@link #DEFAULT_DATA_PAGE_SIZE}.
     *
     * @param arrayDir  directory for array data store
     * @param arrayName the name of the array, will be appended as last part of the array directory
     * @throws IOException exception throws during array initialization
     */
    public BigArrayImpl(String arrayDir, String arrayName) throws IOException {
        this(arrayDir, arrayName, DEFAULT_DATA_PAGE_SIZE);
    }

    /**
     * A big array implementation supporting sequential write and random read.
     *
     * @param arrayDir  directory for array data store
     * @param arrayName the name of the array, will be appended as last part of the array directory
     * @param pageSize  the back data file size per page in bytes, see minimum allowed {@link #MINIMUM_DATA_PAGE_SIZE}.
     * @throws IOException exception throws during array initialization
     */
    public BigArrayImpl(String arrayDir, String arrayName, int pageSize) throws IOException {
        arrayDirectory = arrayDir; // 数据目录
        if (!arrayDirectory.endsWith(File.separator)) {
            arrayDirectory += File.separator;
        }
        // 在一个文件夹下新建一个名为 arrayName 的文件夹存储数据¬
        // append array name as part of the directory
        arrayDirectory = arrayDirectory + arrayName + File.separator;

        // validate directory
        if (!FileUtil.isFilenameValid(arrayDirectory)) {
            throw new IllegalArgumentException("invalid array directory : " + arrayDirectory);
        }

        if (pageSize < MINIMUM_DATA_PAGE_SIZE) {
            throw new IllegalArgumentException("invalid page size, allowed minimum is : " + MINIMUM_DATA_PAGE_SIZE + " bytes.");
        }

        DATA_PAGE_SIZE = pageSize;

        this.commonInit();
    }

    public String getArrayDirectory() {
        return this.arrayDirectory;
    }


    void commonInit() throws IOException {
        // initialize page factories
        this.indexPageFactory = new MappedPageFactoryImpl(INDEX_PAGE_SIZE,
                this.arrayDirectory + INDEX_PAGE_FOLDER,
                INDEX_PAGE_CACHE_TTL); // 索引页
        this.dataPageFactory = new MappedPageFactoryImpl(DATA_PAGE_SIZE,
                this.arrayDirectory + DATA_PAGE_FOLDER,
                DATA_PAGE_CACHE_TTL); // 数据页
        // the ttl does not matter here since meta data page is always cached
        this.metaPageFactory = new MappedPageFactoryImpl(META_DATA_PAGE_SIZE,
                this.arrayDirectory + META_DATA_PAGE_FOLDER,
                10 * 1000/*does not matter*/); // 元数据页

        // initialize array indexes
        initArrayIndex(); // 元数据

        // initialize data page indexes
        initDataPageIndex(); // 页数据
    }

    @Override
    public void removeAll() throws IOException {
        try {
            arrayWriteLock.lock();
            this.indexPageFactory.deleteAllPages();
            this.dataPageFactory.deleteAllPages();
            this.metaPageFactory.deleteAllPages();
            //FileUtil.deleteDirectory(new File(this.arrayDirectory));

            this.commonInit(); // 重新初始化
        } finally {
            arrayWriteLock.unlock();
        }
    }

    // 删除序号之前的数据
    @Override
    public void removeBeforeIndex(long index) throws IOException {
        try {
            arrayWriteLock.lock();

            validateIndex(index); // 校验序号，是否越界

            long indexPageIndex = Calculator.div(index, INDEX_ITEMS_PER_PAGE_BITS);

            ByteBuffer indexItemBuffer = this.getIndexItemBuffer(index);
            long dataPageIndex = indexItemBuffer.getLong();

            if (indexPageIndex > 0L) {
                this.indexPageFactory.deletePagesBeforePageIndex(indexPageIndex);
            }
            if (dataPageIndex > 0L) {
                this.dataPageFactory.deletePagesBeforePageIndex(dataPageIndex);
            }

            // advance the tail to index
            this.arrayTailIndex.set(index);
        } finally {
            arrayWriteLock.unlock();
        }
    }


    @Override
    public void removeBefore(long timestamp) throws IOException {
        try {
            arrayWriteLock.lock();
            long firstIndexPageIndex = this.indexPageFactory.getFirstPageIndexBefore(timestamp);
            if (firstIndexPageIndex >= 0) {
//				long nextIndexPageIndex = firstIndexPageIndex;
//				if (nextIndexPageIndex == Long.MAX_VALUE) { //wrap
//					nextIndexPageIndex = 0L;
//				} else {
//					nextIndexPageIndex++;
//				}
                long toRemoveBeforeIndex = Calculator.mul(firstIndexPageIndex, INDEX_ITEMS_PER_PAGE_BITS);
                removeBeforeIndex(toRemoveBeforeIndex);
            }
        } catch (IndexOutOfBoundsException ex) {
            // ignore
        } finally {
            arrayWriteLock.unlock();
        }
    }

    // find out array head/tail from the meta data
    void initArrayIndex() throws IOException {
        IMappedPage metaDataPage = this.metaPageFactory.acquirePage(META_DATA_PAGE_INDEX);
        ByteBuffer metaBuf = metaDataPage.getLocal(0);
        long head = metaBuf.getLong();
        long tail = metaBuf.getLong();

        arrayHeadIndex.set(head);
        arrayTailIndex.set(tail);
    }

    // find out data page head index and offset
    void initDataPageIndex() throws IOException {
        if (this.isEmpty()) { // 如果 array 为空
            headDataPageIndex = 0L;
            headDataItemOffset = 0;
        } else {
            IMappedPage previousIndexPage = null;
            long previousIndexPageIndex = -1;
            try {
                long previousIndex = this.arrayHeadIndex.get() - 1;
                previousIndexPageIndex = Calculator.div(previousIndex, INDEX_ITEMS_PER_PAGE_BITS); // shift optimization
                previousIndexPage = this.indexPageFactory.acquirePage(previousIndexPageIndex); // 获取之前的索引页
                int previousIndexPageOffset = (int) (Calculator.mul(Calculator.mod(previousIndex, INDEX_ITEMS_PER_PAGE_BITS), INDEX_ITEM_LENGTH_BITS));
                ByteBuffer previousIndexItemBuffer = previousIndexPage.getLocal(previousIndexPageOffset);
                long previousDataPageIndex = previousIndexItemBuffer.getLong(); // 页序号
                int previousDataItemOffset = previousIndexItemBuffer.getInt(); // 偏移
                int perviousDataItemLength = previousIndexItemBuffer.getInt(); // 长度

                headDataPageIndex = previousDataPageIndex;
                headDataItemOffset = previousDataItemOffset + perviousDataItemLength;
            } finally {
                if (previousIndexPage != null) {
                    this.indexPageFactory.releasePage(previousIndexPageIndex);
                }
            }
        }
    }

    /**
     * Append the data into the head of the array
     */
    public long append(byte[] data) throws IOException {
        try {
            arrayReadLock.lock(); // 加锁
            IMappedPage toAppendDataPage = null; // 待添加的数据页
            IMappedPage toAppendIndexPage = null; // 待添加的索引页
            long toAppendIndexPageIndex = -1L; // 待添加的索引页 索引项序号
            long toAppendDataPageIndex = -1L; // 待添加的数据页 数据项序号

            long toAppendArrayIndex = -1L; // 待添加的数组序号
            // 由此可见，一条数据被添加时，分别对于的数组、索引页、数据页三个部分

            try {
                appendLock.lock(); // only one thread can append

                // prepare the data pointer
                // 如果头部数据项偏移 + 数据长度 > page_size
                // 即如果添加的数据内容会超过当前页的范围，那么开始新的数据页
                if (this.headDataItemOffset + data.length > DATA_PAGE_SIZE) { // not enough space
                    this.headDataPageIndex++;
                    this.headDataItemOffset = 0;
                }

                toAppendDataPageIndex = this.headDataPageIndex;
                int toAppendDataItemOffset = this.headDataItemOffset;
                // 添加到头部
                toAppendArrayIndex = this.arrayHeadIndex.get();

                // append data
                // 得到数据页
                toAppendDataPage = this.dataPageFactory.acquirePage(toAppendDataPageIndex);
                // 得到待追加的 buffer
                ByteBuffer toAppendDataPageBuffer = toAppendDataPage.getLocal(toAppendDataItemOffset);
                // 添加页数据
                toAppendDataPageBuffer.put(data);
                // 脏页
                toAppendDataPage.setDirty(true);
                // update to next
                // 重新计算下一个数据的偏移量
                this.headDataItemOffset += data.length;
                // 计算得到 索引页 的序号
                toAppendIndexPageIndex = Calculator.div(toAppendArrayIndex, INDEX_ITEMS_PER_PAGE_BITS); // shift optimization
                // 得到索引页
                toAppendIndexPage = this.indexPageFactory.acquirePage(toAppendIndexPageIndex);
                // 得到索引页项偏移
                int toAppendIndexItemOffset = (int) (Calculator.mul(Calculator.mod(toAppendArrayIndex, INDEX_ITEMS_PER_PAGE_BITS), INDEX_ITEM_LENGTH_BITS));

                // update index 更新索引
                ByteBuffer toAppendIndexPageBuffer = toAppendIndexPage.getLocal(toAppendIndexItemOffset);
                toAppendIndexPageBuffer.putLong(toAppendDataPageIndex); // 数据页 ；64 位
                toAppendIndexPageBuffer.putInt(toAppendDataItemOffset); // 数据项偏移； 32 位
                toAppendIndexPageBuffer.putInt(data.length); // 数据项长度；32 位
                long currentTime = System.currentTimeMillis();
                toAppendIndexPageBuffer.putLong(currentTime); // 时间戳；64 位
                toAppendIndexPage.setDirty(true); // 脏页

                // advance the head
                this.arrayHeadIndex.incrementAndGet(); // 头部指针加1

                // update meta data 更新元数据，元数据总共也就 1 页，即第 0 页
                IMappedPage metaDataPage = this.metaPageFactory.acquirePage(META_DATA_PAGE_INDEX);
                ByteBuffer metaDataBuf = metaDataPage.getLocal(0);
                metaDataBuf.putLong(this.arrayHeadIndex.get());
                metaDataBuf.putLong(this.arrayTailIndex.get());
                metaDataPage.setDirty(true);

            } finally {

                appendLock.unlock();
                // 释放数据页和索引页
                if (toAppendDataPage != null) {
                    this.dataPageFactory.releasePage(toAppendDataPageIndex);
                }
                if (toAppendIndexPage != null) {
                    this.indexPageFactory.releasePage(toAppendIndexPageIndex);
                }
            }

            return toAppendArrayIndex;

        } finally {
            arrayReadLock.unlock();
        }
    }

    @Override
    public void flush() {
        try {
            arrayReadLock.lock();

//			try {
//				appendLock.lock(); // make flush and append mutually exclusive

            this.metaPageFactory.flush();
            this.indexPageFactory.flush();
            this.dataPageFactory.flush();

//			} finally {	
//				appendLock.unlock();
//			}

        } finally {
            arrayReadLock.unlock();
        }

    }

    public byte[] get(long index) throws IOException {
        try {
            arrayReadLock.lock();
            validateIndex(index);

            IMappedPage dataPage = null;
            long dataPageIndex = -1L;
            try {
                // 通过索引拿到索引页数据
                ByteBuffer indexItemBuffer = this.getIndexItemBuffer(index);
                // 索引项存储的第一个是数据页的索引
                dataPageIndex = indexItemBuffer.getLong();
                // 第二个是数据项的偏移量
                int dataItemOffset = indexItemBuffer.getInt();
                // 第三个是数据项的长度
                int dataItemLength = indexItemBuffer.getInt();
                // 获取数据页
                dataPage = this.dataPageFactory.acquirePage(dataPageIndex);
                // 通过偏移量、长度拿到数据体
                byte[] data = dataPage.getLocal(dataItemOffset, dataItemLength);
                return data; // 返回数据
            } finally {
                if (dataPage != null) { // 释放索引页
                    this.dataPageFactory.releasePage(dataPageIndex);
                }
            }
        } finally {
            arrayReadLock.unlock();
        }
    }

    public long getTimestamp(long index) throws IOException {
        try {
            arrayReadLock.lock();
            validateIndex(index);

            ByteBuffer indexItemBuffer = this.getIndexItemBuffer(index);
            // position to the timestamp
            int position = indexItemBuffer.position();
            indexItemBuffer.position(position + INDEX_ITEM_DATA_ITEM_TIMESTAMP_OFFSET);
            long ts = indexItemBuffer.getLong();
            return ts;
        } finally {
            arrayReadLock.unlock();
        }
    }

    ByteBuffer getIndexItemBuffer(long index) throws IOException {

        IMappedPage indexPage = null;
        long indexPageIndex = -1L;
        try {
            // 计算索引页的序号，这里 INDEX_ITEMS_PER_PAGE_BITS 大小为 17
            // 通过右移运算，如果 index = 0，那么 indexPageIndex = 0
            // 如果 index = 1，那么 indexPageIndex = 1 >> 17 = 0
            // 因此一个索引页的大小是固定的，索引项的大小也是固定的，索引项的个数更是固定的
            // 索引项的个数为 1 << 17 = 131072 个
            // 1024 * 128 每一个索引页大小位 1024 * 128 bit
            // 每一个索引项 128 bit
            indexPageIndex = Calculator.div(index, INDEX_ITEMS_PER_PAGE_BITS); // shift optimization
            // 获得索引页
            indexPage = this.indexPageFactory.acquirePage(indexPageIndex);
            // 计算 索引项 所在的偏移量
            // 此处 mod 处理是为了向下取整，从而得到索引页的序号
            // 索引项的大小是 32 位，这个地方相当于 32 * (index / 2^17)
            // 先计算出 数组索引 所对应的 索引项序号
            // 然后计算出 索引项序号 对应的 索引项偏移
            int indexItemOffset = (int) (Calculator.mul(Calculator.mod(index, INDEX_ITEMS_PER_PAGE_BITS), INDEX_ITEM_LENGTH_BITS));
            // 得到 offset 处的 buffer
            ByteBuffer indexItemBuffer = indexPage.getLocal(indexItemOffset);
            return indexItemBuffer; // 返回 buffer
        } finally {
            if (indexPage != null) { // 使用完毕，记得释放
                this.indexPageFactory.releasePage(indexPageIndex);
            }
        }
    }

    // 检查索引
    void validateIndex(long index) {
        // 如果尾部索引小于头部，数组其实是个环
        if (this.arrayTailIndex.get() <= this.arrayHeadIndex.get()) {
            // 小于尾部 或者 大于头部，都是错误索引
            if (index < this.arrayTailIndex.get() || index >= this.arrayHeadIndex.get()) {
                throw new IndexOutOfBoundsException();
            }
        } else {
            // 另外的情况，索引如果小于尾部且大于头部，则索引错误，即不能在二者之间
            if (index < this.arrayTailIndex.get() && index >= this.arrayHeadIndex.get()) {
                throw new IndexOutOfBoundsException();
            }
        }
    }

    public long size() {
        try {
            arrayReadLock.lock();
            return (this.arrayHeadIndex.get() - this.arrayTailIndex.get());
        } finally {
            arrayReadLock.unlock();
        }
    }

    public long getHeadIndex() {
        try {
            arrayReadLock.lock();
            return arrayHeadIndex.get();
        } finally {
            arrayReadLock.unlock();
        }
    }

    public long getTailIndex() {
        try {
            arrayReadLock.lock();
            return arrayTailIndex.get();
        } finally {
            arrayReadLock.unlock();
        }
    }

    @Override
    public boolean isEmpty() {
        try {
            arrayReadLock.lock();
            return this.arrayHeadIndex.get() == this.arrayTailIndex.get();
        } finally {
            arrayReadLock.unlock();
        }
    }

    @Override
    public boolean isFull() {
        // full means the java long space has been used up,
        // the end of the world:)
        return false;
    }

    @Override
    public void close() throws IOException {
        try {
            arrayWriteLock.lock();
            if (this.metaPageFactory != null) {
                this.metaPageFactory.releaseCachedPages();
            }
            if (this.indexPageFactory != null) {
                this.indexPageFactory.releaseCachedPages();
            }
            if (this.dataPageFactory != null) {
                this.dataPageFactory.releaseCachedPages();
            }
        } finally {
            arrayWriteLock.unlock();
        }
    }

    @Override
    public int getDataPageSize() {
        return DATA_PAGE_SIZE;
    }

    // 寻找最接近时间戳的数组索引，注意是数组的索引
    @Override
    public long findClosestIndex(long timestamp) throws IOException {
        try {
            arrayReadLock.lock();
            long closestIndex = NOT_FOUND;
            long tailIndex = this.arrayTailIndex.get();
            long headIndex = this.arrayHeadIndex.get();
            if (tailIndex == headIndex) return closestIndex; // empty
            long lastIndex = headIndex - 1;
            if (lastIndex < 0) {
                lastIndex = Long.MAX_VALUE;
            }
            if (tailIndex < lastIndex) {
                closestIndex = closestBinarySearch(tailIndex, lastIndex, timestamp);
            } else { // 在圆环的另一侧，数组是个环
                long lowPartClosestIndex = closestBinarySearch(0L, lastIndex, timestamp);
                long highPartClosetIndex = closestBinarySearch(tailIndex, Long.MAX_VALUE, timestamp);

                long lowPartTimestamp = this.getTimestamp(lowPartClosestIndex);
                long highPartTimestamp = this.getTimestamp(highPartClosetIndex);

                closestIndex = Math.abs(timestamp - lowPartTimestamp) < Math.abs(timestamp - highPartTimestamp)
                        ? lowPartClosestIndex : highPartClosetIndex;
            }

            return closestIndex;
        } finally {
            arrayReadLock.unlock();
        }
    }

    private long closestBinarySearch(long low, long high, long timestamp) throws IOException {
        long mid;
        long sum = low + high;
        if (sum < 0) { // overflow 溢出
            BigInteger bigSum = BigInteger.valueOf(low);
            bigSum = bigSum.add(BigInteger.valueOf(high));
            mid = bigSum.shiftRight(1).longValue();
        } else {
            mid = sum / 2;
        }
        // 拿到中点的时间戳
        long midTimestamp = this.getTimestamp(mid);
        // 判断时间戳，典型的二分
        if (midTimestamp < timestamp) {
            long nextLow = mid + 1;
            if (nextLow >= high) { // 已经不能再高了
                return high;
            }
            return closestBinarySearch(nextLow, high, timestamp);
        } else if (midTimestamp > timestamp) {
            long nextHigh = mid - 1;
            if (nextHigh <= low) { // 已经不能再低了
                return low;
            }
            return closestBinarySearch(low, nextHigh, timestamp);
        } else {
            return mid; // 相等直接返回
        }
    }

    @Override
    public long getBackFileSize() throws IOException {
        try {
            arrayReadLock.lock();

            return this._getBackFileSize();

        } finally {
            arrayReadLock.unlock();
        }
    }

    // 计算 back file 大小，然后删除多余的
    @Override
    public void limitBackFileSize(long sizeLimit) throws IOException {
        if (sizeLimit < INDEX_PAGE_SIZE + DATA_PAGE_SIZE) {
            return; // ignore, one index page + one data page are minimum for big array to work correctly
        }

        long backFileSize = this.getBackFileSize();
        if (backFileSize <= sizeLimit) return; // nothing to do

        long toTruncateSize = backFileSize - sizeLimit;
        if (toTruncateSize < DATA_PAGE_SIZE) { // 小于一页
            return; // can't do anything
        }

        try {
            arrayWriteLock.lock();

            // double check
            backFileSize = this._getBackFileSize();
            if (backFileSize <= sizeLimit) return; // nothing to do

            toTruncateSize = backFileSize - sizeLimit;
            if (toTruncateSize < DATA_PAGE_SIZE) {
                return; // can't do anything
            }

            long tailIndex = this.arrayTailIndex.get();
            long headIndex = this.arrayHeadIndex.get();
            long totalLength = 0L;
            while (true) { // 二分找到可以删除的尾页
                if (tailIndex == headIndex) break;
                totalLength += this.getDataItemLength(tailIndex); // 拿到数据项的长度
                if (totalLength > toTruncateSize) break;

                tailIndex++; // 继续向前

                // 如果 mod == 0，证明索引页也会进入下一页，因此加上索引页大小
                if (Calculator.mod(tailIndex, INDEX_ITEMS_PER_PAGE_BITS) == 0) { // take index page into account
                    totalLength += INDEX_PAGE_SIZE; // 也要计算索引页的大小
                }
            }
            // 删除之前的数据，狭义上的说是 0～tailIndex 上的页
            this.removeBeforeIndex(tailIndex);
        } finally {
            arrayWriteLock.unlock();
        }

    }

    @Override
    public int getItemLength(long index) throws IOException {
        try {
            arrayReadLock.lock();
            validateIndex(index);

            return getDataItemLength(index);

        } finally {
            arrayReadLock.unlock();
        }
    }

    private int getDataItemLength(long index) throws IOException {

        ByteBuffer indexItemBuffer = this.getIndexItemBuffer(index);
        // position to the data item length
        int position = indexItemBuffer.position();
        // 重新定位到 length 处， 8 + 4 = 12
        indexItemBuffer.position(position + INDEX_ITEM_DATA_ITEM_LENGTH_OFFSET);
        int length = indexItemBuffer.getInt();
        return length;
    }

    // inner getBackFileSize
    private long _getBackFileSize() throws IOException {
        return this.indexPageFactory.getBackPageFileSize() + this.dataPageFactory.getBackPageFileSize();
    }
}
